package parallel;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import event.Basic;
import event.BasicConfig;
import event.Generator;
import event.PageGenerator;

public class ParallelPageGenerator extends Configured implements Tool {
	static int getNumberOfRegions(JobContext context) {
		Configuration conf = context.getConfiguration();
		return conf.getInt("mapred.map.tasks", 1);
	}
	static String getType(JobContext context) {
		Configuration conf = context.getConfiguration();
	    return conf.get("Basic-type");
	}
	static void setType(Configuration conf, String type) {
		conf.set("Basic-type", type);
	}
	public static class NodeGenMapper extends Mapper<LongWritable,NullWritable,Text,NullWritable> {
		private Generator generator = null;
		private MultipleOutputs<Text,NullWritable> multipleOutputs = null;
		protected void setup(Context context)
		throws IOException, InterruptedException {
			// pre-store the link information for region mode and single level mode
			multipleOutputs = new MultipleOutputs<Text,NullWritable>(context);
			BasicRangeInputSplit inputSplit = (BasicRangeInputSplit)context.getInputSplit();
			// the number of regions will be same as the mapper number
			int numOfRegions = getNumberOfRegions(context);
			String type = getType(context);
			generator = new PageGenerator(multipleOutputs, inputSplit, numOfRegions, type);
			
		}
		protected void map(LongWritable sequence,NullWritable ignored,
				Context context) throws IOException, 
				InterruptedException{
			generator.generate(sequence.get(), "");
		}
		protected void cleanup(Context context)
			throws IOException, InterruptedException {
			multipleOutputs.close();
		}
	}
	/**
	 * @param args the cli arguments
	 * @throws ClassNotFoundException 
	 * @throws InterruptedException 
	 * @throws URISyntaxException 
	 */
	public int run(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
		Configuration conf = getConf();
		setType(conf, args[1]);
		Job job = new Job(conf, "parallelNodeGeneration");
		job.setMapperClass(NodeGenMapper.class);
		job.setNumReduceTasks(0);
		job.setJarByClass(ParallelPageGenerator.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		job.setInputFormatClass(BasicRangeInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextOutputFormat.setOutputPath(job, new Path(args[0] + "/" + args[1]));
		return job.waitForCompletion(false)? 0: 1;
	}

	public static void main(String[] args) throws Exception {
		BasicConfig basicConfig = new BasicConfig();
		List<Basic> basicList = basicConfig.getBasicList();
		// launch a map reduce job for the generation of each page type
		for(int i = 0; i < basicList.size(); i++)
		{
			Basic basic = basicList.get(i);
			String [] params = new String[args.length + 1];
			for(int j = 0; j < args.length; j++)
			{
				params[j] = args[j];
			}
			params[args.length] = basic.getName();
			ToolRunner.run(new Configuration(), new ParallelPageGenerator(), params);
		} 
		//ToolRunner.run(new Configuration(), new ParallelNodeGenerator(), args);
		System.exit(0);
	}

}
